package com.youth.mq.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.codingapi.txlcn.tc.annotation.DTXPropagation;
import com.codingapi.txlcn.tc.annotation.TxcTransaction;
import com.youth.entity.po.Order;
import com.youth.entity.po.OrderItem;
import com.youth.entity.po.OrderShipping;
import com.youth.entity.vo.OrderVO;
import com.youth.feign.commodity.CommodityFeign;
import com.youth.mapper.IOrderItemMapper;
import com.youth.mapper.IOrderMapper;
import com.youth.mapper.IOrderShippingMapper;
import com.youth.utils.OrderUtils;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.Random;

/**
 * @ClassName OrderKafkaConsumer
 * @Description: TODO 商品添加方法的队列消费者
 * @Author yjwu
 * @Date 2020/1/16 16:23
 **/
@Component
@Slf4j
public class OrderKafkaConsumer {
    @Value("${orderToCommodityKafkaTopicName}")
    private String tempKafkaTopicName;

    @Autowired
    private IOrderMapper iorderMapper;
    @Autowired
    private IOrderItemMapper iOrderItemMapper;
    @Autowired
    private IOrderShippingMapper iOrderShippingMapper;
    @Autowired
    private IOrderMapper orderMapper;
    @Autowired
    private CommodityFeign commodityFeign;

    //如果多个集群的情况下，需要在KafkaListener监听注解上添加containerFactory，对应配置中的监听容器工厂。
    @KafkaListener(topics = {"ordertocommodity"})
    @Transactional(rollbackFor = RuntimeException.class)
    @TxcTransaction(propagation = DTXPropagation.SUPPORTS)
    public void receive(ConsumerRecord consumerRecord) {
        try {
            String value = (String) consumerRecord.value();
            log.info("kafka receive msg:{}" + value);
            if (StringUtils.isNotEmpty(value)) {
                OrderVO orderVO;
                orderVO = JSONObject.toJavaObject(JSON.parseObject(value), OrderVO.class);

                Random random = new Random();
                int num = random.nextInt(89) + 10;
                String orderId = orderVO.getUserId() + String.valueOf(System.currentTimeMillis()) + num;
                orderVO.setOrderId(orderId);

                Order order = OrderUtils.orderVoToOrder(orderVO);
                order.setCreateTime(new Date());

                orderVO.getOrderItemVO().setOrderId(orderId);
                OrderItem orderItem = OrderUtils.orderItemVOToOrderItem(orderVO.getOrderItemVO());

                orderVO.getOrderShippingVO().setOrderId(orderId);
                OrderShipping orderShipping = OrderUtils.orderShippingVOToOrderShipping(orderVO.getOrderShippingVO());
                orderShipping.setCreateTime(new Date());

                iorderMapper.insert(order);
                iOrderItemMapper.insert(orderItem);
                iOrderShippingMapper.insert(orderShipping);
                commodityFeign.decreaseStockNumber(orderVO.getOrderItemVO().getTCId(), orderVO.getOrderItemVO().getNum());
            }
        } catch (Exception e) {
            log.info("e.getmsg:" + e.getMessage());
            throw new RuntimeException("kafka throws eception");
        }

    }
}
